package alfred.mdp.worker;

import alfred.mdp.Majordomo;
import alfred.mdp.messages.worker.Disconnect;
import alfred.mdp.messages.worker.Heartbeat;
import alfred.mdp.messages.worker.Ready;
import alfred.mdp.messages.worker.Request;
import alfred.mdp.utils.Shutdowner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZContext;
import org.zeromq.ZDispatcher;
import org.zeromq.ZFrame;
import org.zeromq.ZMQ;
import org.zeromq.ZMsg;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static alfred.mdp.messages.ProtocolMessage.get;
import static alfred.mdp.utils.Utils.asByteArray;

/**
 */
public class Worker {
    private static final Logger LOG = LoggerFactory.getLogger(Worker.class);
    private static final String DEFAULT_LOG_MESSAGE = "Worker {} for service {} - ";

    private final ExecutorService executor = Executors.newCachedThreadPool();
    private final WorkerService service;
    private final ZMQ.Socket socket;
    private final String id;

    private volatile int currentLiveness;
    private final ZDispatcher dispatcher;
    private final ZDispatcher.ZSender sender;

    public Worker(ZContext context, ZDispatcher dispatcher, String endpoint, WorkerService service) {
        this.id = UUID.randomUUID().toString();
        this.service = service;
        this.socket = context.createSocket(ZMQ.DEALER);
        this.socket.setIdentity(asByteArray(id));
        this.socket.connect(endpoint);
        this.dispatcher = dispatcher;
        this.sender = new ZDispatcher.ZSender();
        dispatcher.registerHandler(this.socket, new WorkerHandler(), this.sender);
        ready();//TODO throw exception?
    }

    public void ready() {
        if (LOG.isDebugEnabled()) {
            LOG.debug(DEFAULT_LOG_MESSAGE + "send ready", id, service.getName());
        }
        //TODO:removed if because the send ZMsg method doesn't return a boolean
        sender.send(new Ready(service.getName()).getZMsg());
        resetLiveness();
        Shutdowner.add(getShutdownGroup(), executor.submit(new HeartbeatTask(this)));
    }

    public void sendHb() {
        if (LOG.isDebugEnabled()) {
            LOG.debug(DEFAULT_LOG_MESSAGE + "send heartbeat, liveness is {}", new Object[]{id, service.getName(), currentLiveness});
        }
        currentLiveness--;
        sender.send(new Heartbeat().getZMsg());
        if (currentLiveness <= 0) {
            executor.submit(new ShutdownTask(this, true));
            //TODO:restart hb?
        }
    }

    public void resetLiveness() {
        currentLiveness = Majordomo.HB_LIVENESS;
    }

    public void shutdown(boolean sendDisconnect) {
        if (LOG.isDebugEnabled()) {
            LOG.debug(DEFAULT_LOG_MESSAGE + "shutdown", id, service.getName());
        }
        if (sendDisconnect) {
            sender.send(new Disconnect().getZMsg());
        }
        Shutdowner.shutdown(getShutdownGroup());
        dispatcher.unregisterHandler(socket);
        LOG.info(DEFAULT_LOG_MESSAGE + "shutdown completed", id, service.getName());
    }

    public void shutdown() {
        shutdown(true);
    }

    public String getShutdownGroup() {
        return getClass().getName() + id;
    }

    public WorkerService getService() {
        return service;
    }

    public ExecutorService getExecutor() {
        return executor;
    }

    public String getId() {
        return id;
    }

    private class WorkerHandler implements ZDispatcher.ZMessageHandler {

        @Override
        public void handleMessage(ZDispatcher.ZSender sender, ZMsg received) {
            if (LOG.isTraceEnabled()) {
                LOG.trace(DEFAULT_LOG_MESSAGE + "received message", getId(), getService().getName());
            }
            resetLiveness();

            ZFrame type = get(received, 2);
            if (type.equals(Majordomo.Worker.REQUEST)) {
                Request receivedRequest = new Request(received);
                sender.send(getService().execute(receivedRequest).getZMsg());
                if (LOG.isDebugEnabled()) {
                    LOG.debug(DEFAULT_LOG_MESSAGE + "received request", getId(), getService().getName());
                }

            } else if (type.equals(Majordomo.Worker.HEARTBEAT)) {
                if (LOG.isTraceEnabled()) {
                    LOG.trace(DEFAULT_LOG_MESSAGE + "received heartbeat do nothing current liveness updated on all messages.", getId(), getService().getName());
                }

            } else if (type.equals(Majordomo.Worker.DISCONNECT)) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug(DEFAULT_LOG_MESSAGE + "received disconnect", getId(), getService().getName());
                }
                getExecutor().submit(new ShutdownTask(Worker.this, false));

            } else {
                LOG.error("Attention this message is unknown");

            }
        }
    }
}
